package com.wang.helloworld.event;

import java.util.Collection;
import java.util.EventListener;
import java.util.EventObject;
import java.util.HashSet;
import java.util.Iterator;

import kafka.message.MessageAndMetadata;

public class EventManager {
	private Collection<MessageEventListener> listeners;
	
	public void addMessageListener(MessageEventListener listener) {
        if (listeners == null) {
            listeners = new HashSet<MessageEventListener>();
        }
        listeners.add(listener);
    }
	public void removeMessageListener(MessageEventListener listener) {
        if (listeners == null)
            return;
        listeners.remove(listener);
    }
	protected void notifyListeners(MessageEvent event) {
        Iterator<MessageEventListener> iter = listeners.iterator();
        while (iter.hasNext()) {
        	MessageEventListener listener = (MessageEventListener) iter.next();
            listener.receiveEvent(event);
        }
    }

	
	public class MessageEvent extends EventObject{
		private static final long serialVersionUID = 7058797155181337370L;
		MessageAndMetadata<byte[], byte[]> messageAndMetadata;
		
		public MessageEvent(MessageAndMetadata<byte[], byte[]> source) {
			super(source);
			messageAndMetadata = source;
		}

		public MessageAndMetadata<byte[], byte[]> getEvent() {
			return messageAndMetadata;
		}
	}

	public interface MessageEventListener extends EventListener{
		public void receiveEvent(MessageEvent event);
	}
}
